package com.learn.app;

import com.learn.util.RowUtil;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static com.learn.util.FlinkSinkUtil.getDorisSink_SQL;

public class flinkDemo {

    public static void main(String[] args) {

        String mode = args[0];

        System.setProperty("HADOOP_USER_NAME","root");

        //todo 最关键的步骤！！！，设置为批处理方式
        //todo 最关键的步骤！！！，设置为批处理方式
        //todo 最关键的步骤！！！，设置为批处理方式
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inBatchMode()
                .build();
        TableEnvironment tEnv = TableEnvironment.create(settings);

        //todo  1、连接hive
        RowUtil.setHiveConnect(tEnv,"test_dwm_bty_ds",mode);

        //todo 2、读取hive内的数据
        Table table = tEnv.sqlQuery(
                "SELECT  " +
                        " vin," +
                        " energysyscode ," +
                        " cast(voltage as double)  voltage ," +
                        " cast(`current` as double) `current` ," +
                        " cast(soc as smallint) soc," +
                        " cast(mileage as double ) mileage ," +
                        " cast(command as smallint) command," +
                        " company," +
                        " cast(sampletime as bigint) sampletime " +
                        " FROM test_dwm_bty_ds.dwm_bty_ds_gb_all_company " +
                        " order by sampletime"
        );

        //todo 第二个关键步骤：
        StatementSet statementSet = tEnv.createStatementSet();

        table.printSchema();

        tEnv.executeSql("drop table if exists test");

        tEnv.executeSql("create table test(" +
                " vin string," +
                " energysyscode string," +
                " voltage double," +
                " `current` double," +
                " soc smallint," +
                " mileage double," +
                " command smallint," +
                " company string," +
                " sampletime bigint" +
                ")" + getDorisSink_SQL("test_repair")
        );

        statementSet.addInsert("test",table);

        statementSet.execute();



    }

}
